package com.hrvsr.scheduler.distributed;

import com.hrvsr.common.Request;
import com.hrvsr.common.util.SerializationUtil;
import com.hrvsr.scheduler.DuplicateRemover;
import com.hrvsr.scheduler.Scheduler;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;

public class RedisScheduler implements Scheduler {
	private String queue;
	private int dbNo;
	private JedisPool jedisPool;

	private DuplicateRemover duplicateRemover = null;

	public DuplicateRemover getDuplicateRemover() {
		return duplicateRemover;
	}

	public void setDuplicateRemover(DuplicateRemover duplicateRemover) {
		this.duplicateRemover = duplicateRemover;
	}

	private Jedis getJedis() {
		Jedis jedis = jedisPool.getResource();
		jedis.select(dbNo);
		return jedis;
	}

	@Override
	public Request poll() {
		Jedis jedis = getJedis();
		try {
			byte[] bytes = jedis.rpop(queue.getBytes());
			return SerializationUtil.unSerialize(bytes, Request.class);
		} finally {
			jedis.close();
		}
	}

	@Override
	public void putEvenDuplicate(Request request) {
		byte[] bytes = SerializationUtil.serialize(request);
		Jedis jedis = getJedis();
		try {
			jedis.lpush(queue.getBytes(), bytes);
		} finally {
			jedis.close();
		}
	}

	@Override
	public void offer(Request request) {
		if (duplicateRemover.contains(request))
			return;
		duplicateRemover.add(request);
		putEvenDuplicate(request);
	}

	@Override
	public long size() {
		Jedis jedis = getJedis();
		try {
			return jedis.llen(queue.getBytes());
		} finally {
			jedis.close();
		}
	}

	public String getQueue() {
		return queue;
	}

	public void setQueue(String queue) {
		this.queue = queue;
	}

	public int getDbNo() {
		return dbNo;
	}

	public void setDbNo(int dbNo) {
		this.dbNo = dbNo;
	}

	public JedisPool getJedisPool() {
		return jedisPool;
	}

	public void setJedisPool(JedisPool jedisPool) {
		this.jedisPool = jedisPool;
	}
	
	

}
